package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"mq/config"
	"mq/utils"
	"time"
)

func main() {
	c := utils.Myconsumer
	p := utils.Reproducer

	longility := int64(100 * config.RATIO * config.N)

	ticker := time.NewTicker(time.Millisecond * time.Duration(longility))

	data := utils.NewSyncMap(config.MAX, config.MIN, config.N)

	c.Subscribe(config.NORMALITY_TROPIC, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for i := range msgs {
			data.Append(int(utils.BytesToInt64(msgs[i].Body)), msgs[i].QueueOffset)
			fmt.Println(msgs[i].QueueOffset)
		}
		return consumer.ConsumeSuccess, nil
	})

	c.Start()

	for range ticker.C {
		data.Count(config.MIN, config.MAX)
		utils.SendSyncMessage(config.NORMALITY_DEALED_TROPIC, data.GetDateByByte(), p)
	}
}
